package com.yl.flink.processor;

import com.yl.constant.Const;
import com.yl.entity.MultiDataEntity;
import com.yl.entity.cdc.SettingsDataQuota;
import com.yl.util.FlinkUtil;
import com.yl.util.SUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Optional;

/**
 * @author wlf
 * @since 2022/8/25
 */
public class TagStreamFunc extends ProcessFunction<MultiDataEntity, MultiDataEntity> {

    /**
     * 按照不同的计算类型拆分流
     *
     * @param multiDataEntity 流元素
     * @param ctx             上下文
     * @param out             输出器
     */
    @Override
    public void processElement(MultiDataEntity multiDataEntity, Context ctx, Collector<MultiDataEntity> out) throws Exception {
        for (SettingsDataQuota quota : multiDataEntity.getQuotas()) {
            Optional
                    .ofNullable(quota.getCal_type_code())
                    .ifPresent(calTypeCode -> {
                        // 设置计算类型
                        multiDataEntity.setCalTypeCode(calTypeCode);
                        // 设置计算数据的指标，数据默认0，后面计算后把数据补充上
                        multiDataEntity.setCalValues(Tuple2.of(quota.getCode(), 0d));
                        // 每种计算类型输出一道测流
                        String calKey = getCalKey(quota);
                        multiDataEntity.setCalKey(calKey);
                        ctx.output(FlinkUtil.getTag(calKey), multiDataEntity);
                    });
        }
    }

    /**
     * 获取计算类型唯一标识
     */
    public String getCalKey(SettingsDataQuota quota) {
        String calParams = quota.getCal_params();
        calParams = null != calParams ? calParams : Const.S_NULL;
        return SUtil.fmt(quota.getCal_type_code(), quota.getCal_refers(), calParams);
    }

}
